package com.bins.cloud.stream.controller;

import com.bins.cloud.stream.eo.mq.MqMsg;
import com.bins.cloud.stream.mq.StreamConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Slf4j
@RestController
public class MQController {

    public static final String TOPIC = "TopicTest";

    public static final String TAG = "TagA";
    public static final String[] tags = new String[] { "TagA", "TagB", "TagC", "TagD", "TagE" };

    @Autowired
    private StreamBridge streamBridge;

    @GetMapping("/test")
    public void sendMsg() {
        MqMsg mqMsg = new MqMsg("hello", "world");
        log.info("sendMsg 发送一条记录：" + mqMsg);
        Message<MqMsg> msg = MessageBuilder.withPayload(mqMsg).build();
        streamBridge.send(StreamConstant.PRODUCER_MESSAGE_OUTPUT, mqMsg);
    }

    @PostMapping("/sendMessage")
    public String sendMessage(@RequestParam(value = "message") String message) {
        Map<String, Object> headers = new HashMap<String, Object>();
        headers.put(MessageConst.PROPERTY_KEYS, "test");
        headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, UUID.randomUUID());
        Message<MqMsg> msg = new GenericMessage<MqMsg>(new MqMsg(UUID.randomUUID().toString(), message), headers);
        streamBridge.send(StreamConstant.PRODUCER_MESSAGE_OUTPUT, msg);
        return message;
    }

    @PostMapping("/sendDelayMessage")
    public String sendDelayMessage() {
        MqMsg mqMsg = new MqMsg(UUID.randomUUID().toString(), String.valueOf(System.currentTimeMillis()));
        Message<MqMsg> message = MessageBuilder.withPayload(mqMsg).setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 2).build();
        streamBridge.send(StreamConstant.PRODUCER_MESSAGE_OUTPUT, message);
        return "SUCCESS";
    }

    @PostMapping("/sendRetryMessage")
    public String sendRetryMessage() {
        MqMsg mqMsg = new MqMsg(UUID.randomUUID().toString(), "重试消息");
        Message<MqMsg> message = MessageBuilder.withPayload(mqMsg).build();
        streamBridge.send(StreamConstant.RETRY_MESSAGE_OUTPUT, message);
        return "SUCCESS";
    }

    @PostMapping("/send/msg")
    public String hello(@RequestParam(name = "msg", defaultValue = "hello world") String msg) {
        for (int i = 0; i < 100; i++) {
            String key = "KEY" + i;
            Map<String, Object> headers = new HashMap<>();
            headers.put(MessageConst.PROPERTY_KEYS, key);
            headers.put(MessageConst.PROPERTY_TAGS, tags[i % tags.length]);
            headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i);
            Message<MqMsg> message = new GenericMessage(
                    new MqMsg(String.valueOf(i), msg + " " + i), headers);
            streamBridge.send(StreamConstant.PRODUCER_MESSAGE_OUTPUT, message);
        }
        return "消息发送成功";
    }

}
